﻿
using System;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;

namespace WalkthroughTDF
{
    static class TestBroadcastBlock
    {
        #region "********************* internal classes *********************"
        sealed class Consumer
        {
            private readonly int m_id;
            private readonly IPropagatorBlock<string, string> m_queue;

            public Consumer(int id, IPropagatorBlock<string, string> queue)
            {
                m_id = id;
                m_queue = queue;
            }

            public ITargetBlock<string> IncomingQueue
            {
                get { return m_queue; }
            }

            public void Start()
            {
                m_queue.ReceiveAsync().ContinueWith(this.OnReceived);
            }

            private void OnReceived(Task<string> prevTask)
            {
                Console.WriteLine("Consumer<{0}> got '{1}'", m_id, prevTask.Result);
                Start();
            }
        }

        sealed class ProdConsTester
        {
            #region "member fields"

            private readonly IPropagatorBlock<string, string> m_queue;
            private readonly int m_totalItems;
            private readonly int m_prodInterval;

            #endregion

            #region "constructor"

            public ProdConsTester(IPropagatorBlock<string, string> queue, int totalItems, int prodInterval)
            {
                m_queue = queue;
                m_totalItems = totalItems;
                m_prodInterval = prodInterval;
            }

            #endregion

            #region "public APIs"

            public void Start()
            {
                for (int index = 0; index < m_totalItems; index++)
                {
                    new Consumer(index, m_queue).Start();
                }

                Produce();
            }

            #endregion

            #region "private helpers"

            private void Produce()
            {
                for (int index = 0; index < m_totalItems; index++)
                {
                    m_queue.Post(string.Format("item{0}", index));
                    Thread.Sleep(m_prodInterval);
                }
                m_queue.Complete();
            }

            #endregion
        }
        #endregion

        private static void TestProdConsume(IPropagatorBlock<string, string> queue, int totalItems, int prodInterval)
        {
            new ProdConsTester(queue, totalItems, prodInterval).Start();
            Console.ReadLine();
        }

        /// <summary>
        /// chekanote: broadcast can only PUSH, should not be PULLED
        /// </summary>
        private static void TestUsingBrdcastBlock()
        {
            // ------------------------------ consumers
            var consumer1 = new Consumer(1, new BufferBlock<string>());
            consumer1.Start();

            var consumer2 = new Consumer(2, new BufferBlock<string>());
            consumer2.Start();

            // ------------------------------ producers
            BroadcastBlock<string> prodQueue = new BroadcastBlock<string>(str => str);
            prodQueue.LinkTo(consumer1.IncomingQueue);
            prodQueue.LinkTo(consumer2.IncomingQueue);

            for (int index = 0; index < 4; index++)
            {
                prodQueue.Post(string.Format("item{0}", index));
                Thread.Sleep(100);
            }//for

            // ------------------------------ hold and wait
            Console.ReadLine();
        }

        public static void TestMain()
        {
            // TestProdConsume(new BufferBlock<string>(), 2, 100);

            // chekanote: !!! because broadcast block is overwrite queue, which always keep the last value
            // so it will never be empty
            // chekanote: !!! this means a broadcast queue can not be pulled, because it can be never empty
            // it always keep the last value so it can never be empty
            // it only be linked to other targets, and actively copy and push data to following targets
            // TestProdConsume(new BroadcastBlock<string>(str => str), 2, 100);

            TestUsingBrdcastBlock();
        }// TestMain
    }
}
